package core

import (
	"context"
	"crypto/tls"
	"fmt"
	"github.com/alitto/pond"
	"github.com/rpcxio/rpcx-etcd/serverplugin"
	"github.com/smallnest/rpcx/server"
	"octopus/hextech/nlog"
	"time"
)

type RpcXServerOptionsPoolConfig struct {
	MaxWorkers  int // worker 工作goroutine的数量
	MaxCapacity int // capacity 最大的待处理的请求 超过capacity，新的请求就会被阻塞。
	PondOptions []pond.Option
}

type RpcXServerEtcdConfig struct {
	Nodes          []string
	BasePath       string
	UpdateInterval time.Duration
	Expired        time.Duration
}

type RpcXServerBaseConfig struct {
	Port       int    // rpc服务端口
	Address    string // rpc服务地址
	ServerName string // rpc服务名
	CertFile   string // tls证书
	KeyFile    string // tls证书
}

type RpcXServerOptionsConfig struct {
	tlsConfig          *tls.Config                  // TLS 配置项
	ReadTimeout        time.Duration                // 读取超时时间
	WriteTimeout       time.Duration                // 写入超时时间
	Pool               *RpcXServerOptionsPoolConfig // 默认pond协程池 alitto/pond
	CustomPool         server.WorkerPool            // 自定义协程池 CustomPool和pool互斥
	AsyncWrite         bool                         // 是否异步写入
	TCPKeepAlivePeriod time.Duration                // tcp保持有效期
}

type RpcXServerConfig struct {
	Base    *RpcXServerBaseConfig
	Options *RpcXServerOptionsConfig
	Etcd    *RpcXServerEtcdConfig
}

type RpcXServer struct {
	Address      string // host:port
	Network      string // tcp
	options      []server.OptionFn
	server       *server.Server
	etcdV3Plugin *serverplugin.EtcdV3RegisterPlugin
	Plugins      []*server.Plugin // 启动rpc服务时，自定义插件
	ServerName   string           // rpc服务名
	Receiver     interface{}      // 路由结构 - 指针
}

func (rpc *RpcXServer) Run() {
	if rpc == nil || rpc.Receiver == nil {
		return
	}
	ctx := context.Background()
	srv := server.NewServer(rpc.options...)
	rpc.server = srv
	// 添加服务发现
	err := rpc.etcdV3Plugin.Start()
	if err != nil {
		nlog.Fatalf(ctx, "Add rpc server %v etcd plugin failed: %v", rpc.ServerName, err.Error())
	}
	rpc.server.Plugins.Add(rpc.etcdV3Plugin)
	// 添加自定义插件
	if len(rpc.Plugins) > 0 {
		rpc.server.Plugins.Add(rpc.Plugins)
	}
	// 注册服务路由
	if err = rpc.server.RegisterName(rpc.ServerName, rpc.Receiver, ""); err != nil {
		nlog.Fatalf(ctx, "Register rpc server %v failed: %v", rpc.ServerName, err.Error())
	}
	go func() {
		nlog.Infof(ctx, "Start to listening the incoming requests on rpc server %v address: %v", rpc.ServerName, rpc.Address)
		err = rpc.server.Serve(rpc.Network, rpc.Address)
		if err != nil {
			nlog.Fatalf(ctx, "Start rpc server %v failed: %v", rpc.ServerName, err.Error())
		}
		nlog.Infof(ctx, "Rpc server %v on %s stopped", rpc.ServerName, rpc.Address)
	}()
}

func (rpc *RpcXServer) Close() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if err := rpc.server.Shutdown(ctx); err != nil {
		nlog.Warnf(context.Background(), "Shutdown secure rpc server %v failed: %s", rpc.ServerName, err.Error())
	}
}

func NewRpcXServer(config *RpcXServerConfig) (*RpcXServer, error) {
	base := config.Base
	opt := config.Options

	if base.CertFile != "" && base.KeyFile != "" {
		if opt == nil {
			opt = new(RpcXServerOptionsConfig)
		}
		cert, _ := tls.LoadX509KeyPair("server.pem", "server.key")
		opt.tlsConfig = &tls.Config{Certificates: []tls.Certificate{cert}}
	}
	var opts []server.OptionFn

	if opt != nil {
		if opt.tlsConfig != nil {
			opts = append(opts, server.WithTLSConfig(opt.tlsConfig))
		}
		if opt.ReadTimeout > 0 {
			opts = append(opts, server.WithReadTimeout(opt.ReadTimeout))
		}
		if opt.WriteTimeout > 0 {
			opts = append(opts, server.WithReadTimeout(opt.WriteTimeout))
		}
		// 如果配置了协程池，则优先使用自定义协程池
		if opt.CustomPool != nil {
			opts = append(opts, server.WithCustomPool(opt.CustomPool))
		} else if opt.Pool != nil {
			opts = append(opts, server.WithPool(opt.Pool.MaxWorkers, opt.Pool.MaxCapacity, opt.Pool.PondOptions...))
		}
		// 注意：官方不推荐异步写入
		if opt.AsyncWrite {
			opts = append(opts, server.WithAsyncWrite())
		}
		// TCP连接默认保持3分钟
		if opt.TCPKeepAlivePeriod > 0 {
			opts = append(opts, server.WithTCPKeepAlivePeriod(opt.TCPKeepAlivePeriod))
		}
	}

	etcd := config.Etcd
	rpcXserver := &RpcXServer{
		Address: base.getAddress(),
		Network: base.getNetwork(),
		options: opts,
		etcdV3Plugin: &serverplugin.EtcdV3RegisterPlugin{
			ServiceAddress: base.getNetwork() + "@" + base.getAddress(),
			EtcdServers:    etcd.Nodes,
			BasePath:       etcd.BasePath,
			UpdateInterval: etcd.getUpdateInterval(),
			Expired:        etcd.getExpired(),
		},
		ServerName: base.ServerName,
	}

	return rpcXserver, nil
}

// 默认使用tcp
func (o *RpcXServerBaseConfig) getNetwork() string {
	return "tcp"
}

// 拼接地址 host:port
func (o *RpcXServerBaseConfig) getAddress() string {
	return fmt.Sprintf("%v:%v", o.Address, o.Port)
}

// etcd更新间隔，默认30s
func (e *RpcXServerEtcdConfig) getUpdateInterval() time.Duration {
	if e.UpdateInterval <= 0 {
		return 30 * time.Second
	}
	return e.UpdateInterval
}

// etcd过期时间，默认2分钟
func (e *RpcXServerEtcdConfig) getExpired() time.Duration {
	if e.Expired <= 0 {
		return 2 * time.Minute
	}
	return e.Expired
}
